package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.offline.accumulator.DCConstants;
import com.navinfo.tripanalysis.common.pojo.Point;
import com.navinfo.tripanalysis.common.pojo.Point0200;
import com.navinfo.tripanalysis.common.util.BigDecimalUtils;
import com.navinfo.tripanalysis.offline.pojo.Abnormal0200DataDetail;
import com.navinfo.tripanalysis.offline.service.DataCleaning0200Service;
import com.navinfo.tripanalysis.offline.util.AccumulatorUtils;
import com.navinfo.tripanalysis.offline.util.DataNode;
import com.navinfo.tripanalysis.offline.util.OutlierNodeDetect;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.util.AccumulatorV2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.*;

import static java.lang.Math.abs;
import static java.lang.Math.floor;
import static java.lang.Math.log;

@Data
public class DataCleaning0200ServiceImpl implements DataCleaning0200Service, Serializable {

    private static final Logger logger = LoggerFactory.getLogger(DataCleaning0200ServiceImpl.class);

    private  double Speed_Max_Threshold;
    private  double Fuel_100km_Max_Threshold;
    private  int Point_Time;
    private  double Mileage_Constant;
    private  double LatLon_Max_Threshold;

    //初始化各统计指标
    //车辆上传记录数
    private int point_count = 0;
    //gpstime重复记录数
    private int repeated_gpstime_count = 0;
    //里程异常记录数
    private int abnormal_mileage_count = 0;
    //油耗异常记录数
    private int abnormal_fuel_count = 0;
    //经纬度漂移记录数
    private int abnormal_lat_lon_count = 0;
    //里程平均速度异常范围
    //速度150-200
    private int mileage_avgspeedRange_150_200_count = 0;
    //速度200-500
    private int mileage_avgspeedRange_200_500_count = 0;
    //速度500
    private int mileage_avgspeedRange_500_count = 0;
    //百公里油耗异常范围
    //油耗150-200
    private int fuelRange_150_200_count = 0;
    //油耗200-500
    private int fuelRange_200_500_count = 0;
    //油耗500以上
    private int fuelRange_500_count = 0;

    //保存结果List
    List<Point0200> resultList ;
    //时间间隔不足60s集合
    List<Point0200> less_60s_points;

    /**
     * 0200数据
     * @param pointsRdd
     * @param accumulator
     * @return
     */
    @Override
    public JavaPairRDD<Long, Tuple2<Abnormal0200DataDetail, List<Point0200>>> getAbnormal0200Data(JavaPairRDD<Long, List<Point>> pointsRdd, AccumulatorV2<String, String> accumulator){

        return pointsRdd.mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, List<Point>>>, Long, Tuple2<Abnormal0200DataDetail, List<Point0200>>>) tuplelist ->
                new Iterator<Tuple2<Long, Tuple2<Abnormal0200DataDetail, List<Point0200>>>>() {
                    @Override
                    public boolean hasNext() {
                        return tuplelist.hasNext();
                    }

                    @Override
                    public Tuple2<Long, Tuple2<Abnormal0200DataDetail, List<Point0200>>> next() {

                        //step1:gpstime 去重
                        Tuple2<Long, List<Point>> tuple = tuplelist.next();
                        Long tid = tuple._1;
                        List<Point0200> points_0200 = (List)tuple._2();
                        point_count = points_0200.size();
                        //判断初始坐标是否发生漂移
                        if(point_count >= 100) {
                            checkFirstPoint(points_0200, Math.min((int) (point_count * 0.01), 100) + 10);
                        }
                        //结果集合
                        resultList = new ArrayList<>();
                        //将第一个点作为，初始对比点
                        Point0200 prepoint = points_0200.get(0);

                        //将第一点插入结果集
                        resultList.add(prepoint);
                        //从第二个点开始计算
                        points_0200.remove(0);
                        //时间间隔不足60s集合
                        less_60s_points = new ArrayList<>();
                        //遍历所有数据，判断异常数值
                        for (Point0200 p : points_0200) {

                            long pretime = prepoint.getGpsTime();
                            if(less_60s_points.size() >= 1 && CommonUtils.isCollectionNotEmpty(less_60s_points)){
                                pretime = less_60s_points.get(less_60s_points.size() - 1).getGpsTime();
                            }
                            //过滤重复点
                            if (p.getGpsTime() == pretime) {
                                ++repeated_gpstime_count;
                                continue;
                            }
                            //acc=0的点不做判断,直接插入结果
                            if (p.getAccStatus() == 1) {
                                //按规则处理每个点数据
                                prepoint = process(p, prepoint, less_60s_points);
                            }else{
                                //如果当前点为熄火状态，则处理上一组点
                                if(CommonUtils.isCollectionNotEmpty(less_60s_points)){
                                    processLastPoint(prepoint,less_60s_points);
                                }
                                resultList.add(p);
                                prepoint = p;
                            }

                        }//for循环结束
                        //如果最后一组点，时间累计小于两点时间间隔阈值，处理最后一组点
                        processLastPoint(prepoint,less_60s_points);

                        //统计累加值
                        if(point_count > 0){
                            accumulator.add(DCConstants.ACC_0200_COUNT);
                        }
                        int repeated_point_ratio = BigDecimalUtils.divide(repeated_gpstime_count*100,point_count,0).intValue();
                        if(repeated_point_ratio > 0){
                            accumulator.add(calculateRepeatedPoint(repeated_point_ratio));
                        }
                        if(abnormal_mileage_count >= 10){
                            accumulator.add(calculateMileage(abnormal_mileage_count));
                        }
                        if(abnormal_fuel_count >= 10){
                            accumulator.add(calculateFuel(abnormal_fuel_count));
                        }
                        if(abnormal_lat_lon_count >= 10){
                            accumulator.add(calculateLatAndLon(abnormal_lat_lon_count));
                        }

                        //汇总异常数据明细
                        Abnormal0200DataDetail data = null;
                        if(abnormal_mileage_count+abnormal_fuel_count > 0 || repeated_gpstime_count >= 100){
                            data = new Abnormal0200DataDetail();
                            data.setTid(tid);
                            data.setPoint_count(point_count);
                            data.setRepeated_gpstime_count(repeated_gpstime_count);
                            data.setMileage_avgspeedRange_150_200_count(mileage_avgspeedRange_150_200_count);
                            data.setMileage_avgspeedRange_200_500_count(mileage_avgspeedRange_200_500_count);
                            data.setMileage_avgspeedRange_500_count(mileage_avgspeedRange_500_count);
                            data.setFuelRange_150_200_count(fuelRange_150_200_count);
                            data.setFuelRange_200_500_count(fuelRange_200_500_count);
                            data.setFuelRange_500_count(fuelRange_500_count);
                            data.setLat_lon_Range_count(abnormal_lat_lon_count);
                        }

                        //初始化各统计指标
                        //车辆上传记录数
                        point_count = 0;
                        //gpstime重复记录数
                        repeated_gpstime_count = 0;
                        //里程异常记录数
                        abnormal_mileage_count = 0;
                        //油耗异常记录数
                        abnormal_fuel_count = 0;
                        //经纬度漂移记录数
                        abnormal_lat_lon_count = 0;
                        //里程平均速度异常范围
                        //速度150-200
                        mileage_avgspeedRange_150_200_count = 0;
                        //速度200-500
                        mileage_avgspeedRange_200_500_count = 0;
                        //速度500
                        mileage_avgspeedRange_500_count = 0;
                        //百公里油耗异常范围
                        //油耗150-200
                        fuelRange_150_200_count = 0;
                        //油耗200-500
                        fuelRange_200_500_count = 0;
                        //油耗500以上
                        fuelRange_500_count = 0;
                        return new Tuple2<>(tid,new Tuple2<>(data,resultList));
                    }
                });
    }//清洗方法结束

    /**
     * 更新累加器里程异常
     * @param abnormal_mileage_count
     * @return
     */
    private String calculateMileage(int abnormal_mileage_count){

        if(abnormal_mileage_count >= 10 && abnormal_mileage_count < 100){
            return DCConstants.ACC_0200_MILEAGE_10_100_COUNT;
        }else if(abnormal_mileage_count >= 100 && abnormal_mileage_count < 300){
            return DCConstants.ACC_0200_MILEAGE_100_300_COUNT;
        }else if(abnormal_mileage_count >= 300){
            return DCConstants.ACC_0200_MILEAGE_300_COUNT;
        }
        return null;
    }

    /**
     * 更新累加器油耗异常
     * @param abnormal_fuel_count
     * @return
     */
    private String calculateFuel(int abnormal_fuel_count){

        if(abnormal_fuel_count >= 10 && abnormal_fuel_count < 100){
            return DCConstants.ACC_0200_FUEL_10_100_COUNT;
        }else if(abnormal_fuel_count >= 100 && abnormal_fuel_count < 300){
            return DCConstants.ACC_0200_FUEL_100_300_COUNT;
        }else if(abnormal_fuel_count >= 300){
            return DCConstants.ACC_0200_FUEL_300_COUNT;
        }
        return null;
    }

    /**
     * 更新累加器IP漂移异常
     * @param abnormal_lat_lon_count
     * @return
     */
    private String calculateLatAndLon(int abnormal_lat_lon_count){
        if(abnormal_lat_lon_count >= 10 && abnormal_lat_lon_count < 100){
            return DCConstants.ACC_0200_LAT_LON_10_100_COUNT;
        }else if(abnormal_lat_lon_count >= 100 && abnormal_lat_lon_count < 300){
            return DCConstants.ACC_0200_LAT_LON_100_300_COUNT;
        }else if(abnormal_lat_lon_count >= 300){
            return DCConstants.ACC_0200_LAT_LON_300_COUNT;
        }
        return null;
    }


    private String calculateRepeatedPoint(int repeated_point_ratio){
        if(repeated_point_ratio > 0 && repeated_point_ratio <= 10){
            return DCConstants.ACC_0200_REPEATED_0_10_Ratio;
        }else if(repeated_point_ratio > 10 && repeated_point_ratio <= 50){
            return DCConstants.ACC_0200_REPEATED_10_50_Ratio;
        }else if(repeated_point_ratio >= 50){
            return DCConstants.ACC_0200_REPEATED_50_Ratio;
        }
        return null;
    }



    /**
     * 判断相邻点的里程、油耗、相邻经纬度距离
     * @param curpoint
     * @param prepoint
     */
    private Point0200 processPoint (Point0200 curpoint,Point0200 prepoint){
        //间隔时长:s
        long diff_gpstime = (curpoint.getGpsTime() - prepoint.getGpsTime())/1000;
        //标准里程:km
        float diff_mileage = curpoint.getStandardMileage() - prepoint.getStandardMileage();
        //标准油耗:L
        float diff_Fuel = curpoint.getStandardFuelCons() - prepoint.getStandardFuelCons();
        //经纬度之间距离:km
        double diff_lat_lon = AccumulatorUtils.getDistance(
                curpoint.getOriginalLat()*Math.pow(10,-6),curpoint.getOriginalLng()*Math.pow(10,-6),
                prepoint.getOriginalLat()*Math.pow(10,-6),prepoint.getOriginalLng()*Math.pow(10,-6));
        //里程计算平均速度:km/h,h=diff_gpstime/3600
        double avgspeed = 3600*diff_mileage/diff_gpstime;
        //百公里油耗：L/100km
        double fuel_100km = diff_mileage==0.0?0.0:100*diff_Fuel/diff_mileage;
        //每小时油耗：L/H
        double fuel_hour = 3600*diff_Fuel/diff_gpstime;

        //判断里程异常（用平均速度表示）
        speedStatistics(avgspeed);
        fuelStatistics(fuel_100km);
        //LatAndLonStatistics(diff_lat_lon,diff_mileage);
        latLonStatistics(curpoint,prepoint,diff_gpstime,avgspeed,diff_mileage);
        directionStatistics(curpoint,prepoint);
        //结果存到List
        if(CommonUtils.isCollectionNotEmpty(less_60s_points)){
            //将临时list中数据加入结果list
            Point0200 last = prepoint;
            for(Point0200 p :less_60s_points){
                //计算多点情况各点按间隔时间指标占比

                long difftime = (p.getGpsTime() - last.getGpsTime())/1000;

                float time = diff_gpstime==0?0:new BigDecimal(difftime).divide(new BigDecimal(diff_gpstime),3,BigDecimal.ROUND_DOWN).floatValue();
                float diffmileage = diff_mileage*time;
                float difffuel = diff_Fuel*time;
                StringBuilder upd = new StringBuilder();
                //不更新初始值
                upd.append(DCConstants.UPDFIELD_STANDARDMILEAGE + "=").append(last.getStandardMileage()+diffmileage)
                        .append("|").append(DCConstants.UPDFIELD_STANDARDFUELCONS + "=").append(last.getStandardFuelCons()+difffuel);
                if(p.getSpeedKmh() > Speed_Max_Threshold){
                    upd.append("|").append(DCConstants.UPDFIELD_SPEEDKMH + "=").append(avgspeed);
                    //p.setSpeedKmh(avgspeed);
                }
                //更新初始值
//                upd.append(DCConstants.UPDFIELD_STANDARDMILEAGE + "=").append(p.getStandardMileage())
//                        .append("|").append(DCConstants.UPDFIELD_STANDARDFUELCONS + "=").append(p.getStandardFuelCons());
//                if(p.getSpeedKmh() > Speed_Max_Threshold){
//                    upd.append("|").append(DCConstants.UPDFIELD_SPEEDKMH + "=").append(p.getSpeedKmh());
//                    p.setSpeedKmh(avgspeed);
//                }
//                p.setStandardMileage(last.getStandardMileage()+diffmileage);
//                p.setStandardFuelCons(last.getStandardFuelCons()+difffuel);
                p.setDataUpd(StringUtils.isEmpty(p.getDataUpd())?upd.toString():p.getDataUpd()+"|"+upd.toString());
                resultList.add(p);
                last = p;
            }
        }
        //当前点插入结果表
        resultList.add(curpoint);
        //清空临时list
        less_60s_points.clear();
        return curpoint;
    }

    /**
     * 当最后一个点与前一个点时间间隔小于阈值时，处理最后一组点
     * @param pre
     * @param less_60s_points
     */
    private void processLastPoint(Point0200 pre,List<Point0200> less_60s_points){
        if (CommonUtils.isCollectionNotEmpty(less_60s_points)){
            Point0200 cur = less_60s_points.get(less_60s_points.size() - 1);
            less_60s_points.remove(less_60s_points.size() - 1);
            processPoint(cur,pre);
        }
    }

    /**
     * 角度为0时根据前后两个经纬度计算方向
     * @param curpoint
     * @param prepoint
     */
    private void directionStatistics(Point0200 curpoint, Point0200 prepoint) {

        //方向为0，修正异常数据
        if(curpoint.getDirection() != 0){return;}
        double lat1 = BigDecimalUtils.divide(curpoint.getOriginalLat(),1000000,6).doubleValue();
        double lon1 = BigDecimalUtils.divide(curpoint.getOriginalLng(),1000000,6).doubleValue();
        double lat2 = BigDecimalUtils.divide(prepoint.getOriginalLat(),1000000,6).doubleValue();
        double lon2 = BigDecimalUtils.divide(prepoint.getOriginalLng(),1000000,6).doubleValue();
        //测试阶段修正值记录到dataUpd列，调整后可将此段代码注释
        String latupd1 = AccumulatorUtils.getFieldFromConcatString(curpoint.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT);
        String lonupd1 = AccumulatorUtils.getFieldFromConcatString(curpoint.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG);
        if(latupd1 != null&&!"".equals(latupd1)){
            lat1 = Double.valueOf(latupd1);
            lon1 = Double.valueOf(lonupd1);
        }
        String latupd2 = AccumulatorUtils.getFieldFromConcatString(prepoint.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT);
        String lonupd2 = AccumulatorUtils.getFieldFromConcatString(prepoint.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG);
        if(latupd2 != null&&!"".equals(latupd2)){
            lat2 = Double.valueOf(latupd2);
            lon2 = Double.valueOf(lonupd2);
        }
        //计算两个坐标之间方向
        double direction = BigDecimal.valueOf(AccumulatorUtils.getDirection(lat1,lon1,lat2,lon2)).setScale(1,BigDecimal.ROUND_DOWN).doubleValue();
        //更新修正值
        String upd = DCConstants.UPDFIELD_0200_DIRECTION+"="+direction;
        curpoint.setDataUpd(StringUtils.isEmpty(curpoint.getDataUpd())?upd:curpoint.getDataUpd()+"|"+upd);
    }

    /**
     * 判断平均速度值是否异常，并按范围区分统计
     * @param speed 速度，KMH
     */
    private void speedStatistics(Double speed){
        if (speed > Speed_Max_Threshold){
            abnormal_mileage_count += (less_60s_points.size()+1);
            //按范围统计异常速度
            setSpeedRange(speed);
        }
    }

    /**
     * 将平均速度按范围统计
     * @param speed
     */
    private void setSpeedRange(Double speed){
        if(speed > 150 && speed <= 200){
            mileage_avgspeedRange_150_200_count += (less_60s_points.size()+1);
        }else if (speed > 200 && speed <= 500){
            mileage_avgspeedRange_200_500_count += (less_60s_points.size()+1);
        }else if (speed > 500){
            mileage_avgspeedRange_500_count += (less_60s_points.size()+1);
        }
    }

    /**
     * 判断百公里油耗是否异常，并按范围区分统计
     * @param fuel
     */
    private void fuelStatistics (Double fuel){
        if(fuel > Fuel_100km_Max_Threshold){
            abnormal_fuel_count += (less_60s_points.size()+1);
            setFuelRange(fuel);
        }
    }

    /**
     * 将百公里油耗按范围统计
     * @param fuel
     */
    private void setFuelRange(Double fuel){
        if(fuel > 150 && fuel <= 200){
            fuelRange_150_200_count += (less_60s_points.size()+1);
        }else if (fuel > 200 && fuel <= 500){
            fuelRange_200_500_count += (less_60s_points.size()+1);
        }else if (fuel > 500){
            fuelRange_500_count += (less_60s_points.size()+1);
        }
    }

    /**
     * 判断是否IP漂移，判断逻辑：
     * 1.两点之间里程为0，经纬度计算距离不为0
     * 2.两点之间里程不为0，经纬度计算距离为0
     * 3.两点之间经纬度计算距离大于里程10倍以上
     * @param diff_lat_lon
     * @param diff_mileage
     */
    private void LatAndLonStatistics (Double diff_lat_lon , Float diff_mileage){
        if((diff_lat_lon >= Mileage_Constant && diff_mileage == 0)
                ||(diff_lat_lon == 0 && diff_mileage >= Mileage_Constant)
                ||(diff_lat_lon > 10*diff_mileage && diff_mileage > 0) ){
            ++abnormal_lat_lon_count;
        }
    }

    private void latLonStatistics (Point0200 cur,Point0200 pre,long time ,double avgSpeed ,float diff_mileage){
        if(pre == null || "".equals(pre.toString())) return;
        //计算经纬度坐标变化差值

        double latLonDiff = time==0?0.0:BigDecimalUtils.divide(getLatLonDiff(cur,pre),time,2).doubleValue();
        if(latLonDiff > LatLon_Max_Threshold){
            abnormal_lat_lon_count += (less_60s_points.size()+1);
            int newLat = 0;
            int newLon = 0;
            int preLat = StringUtils.isEmpty(AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT))?
                    pre.getOriginalLat():
                    Integer.parseInt(AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT));
            int preLon = StringUtils.isEmpty(AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG))?
                    pre.getOriginalLng():
                    Integer.parseInt(AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG));
            if(diff_mileage == 0){
                //经纬度坐标漂移，修正为上一点坐标
                newLat = preLat;
                newLon = preLon;
            }else if(diff_mileage > 0){
                double distance = avgSpeed <= Speed_Max_Threshold? diff_mileage
                        :BigDecimalUtils.divide(time*(cur.getSpeedKmh()+pre.getSpeedKmh()),7200,3).doubleValue();
                Tuple2<Double,Double> newLatLon =
                        AccumulatorUtils.getLatLon(preLat*Math.pow(10,-6), preLon*Math.pow(10,-6),
                                distance,pre.getDirection());
                newLat = BigDecimalUtils.multiply(newLatLon._1,1000000,0).intValue();
                newLon = BigDecimalUtils.multiply(newLatLon._2,1000000,0).intValue();
            }
            //修正结果记录到修正值列
            String updString = DCConstants.UPDFIELD_0200_ORIGINALLAT+"="+newLat+"|"+DCConstants.UPDFIELD_0200_ORIGINALLNG+"="+newLon;
            cur.setDataUpd(StringUtils.isEmpty(cur.getDataUpd()) ?updString :cur.getDataUpd()+"|"+updString);
        }
    }

    /**
     * 判断不同时间间隔位置点处理方式
     * 时间间隔小于60s：将该点加入临时list，直至时间间隔达到小于等于60s的最后一个点，list
     * @param cur
     * @param pre
     * @param
     * @return newpre
     */
    private Point0200 process (Point0200 cur,Point0200 pre,List<Point0200> less_60s_points){
        //返回结果
        Point0200 repoint = pre;
        long type = cur.getGpsTime()-pre.getGpsTime();
        //距上一点时间间隔小于60s，将该点插入小于60slist
        if(type < Point_Time ){
            if((less_60s_points.size() >= 1 && cur.getGpsTime() > less_60s_points.get(less_60s_points.size() -1).getGpsTime() )||less_60s_points.isEmpty()) {
                less_60s_points.add(cur);
            }
        }else if(type == Point_Time && !less_60s_points.isEmpty()){
            //less_60s_points.add(cur);
            repoint = processPoint(cur,pre);
        }else if(type > Point_Time && !less_60s_points.isEmpty()){
            //多个点
            //当前点到临时list之前一个点时间间隔超过60s，分为两个阶段处理
            //1:list最后一个点和prepoint
            //2:cur和list最后一个点
            //取list中最后一个点作为当前点
            Point0200 cur_0 = less_60s_points.get(less_60s_points.size() - 1);
            //在中间list中删除该点
            less_60s_points.remove(cur_0);
            //处理阶段1
            pre = processPoint(cur_0,pre);

            //递归处理阶段2
            repoint = process(cur,pre,less_60s_points);

        }else{//距上一点时间间隔大于60s
            //单个点
            repoint = processPoint(cur, pre);
        }
        return repoint;
    }

    private double getLatLonDiff (Point0200 cur, Point0200 pre){
        if(pre == null){
            return 0;
        }
        String curLatUpd = AccumulatorUtils.getFieldFromConcatString(cur.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT);
        String curLonUpd = AccumulatorUtils.getFieldFromConcatString(cur.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG);
        double curLat = BigDecimalUtils.divide(StringUtils.isEmpty(curLatUpd)?cur.getOriginalLat():Integer.parseInt(curLatUpd),1000000,6).doubleValue();
        double curLon = BigDecimalUtils.divide(StringUtils.isEmpty(curLonUpd)?cur.getOriginalLng():Integer.parseInt(curLonUpd),1000000,6).doubleValue();
        String preLatUpd = AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT);
        String preLonUpd = AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG);
        double preLat = BigDecimalUtils.divide(StringUtils.isEmpty(preLatUpd)?pre.getOriginalLat():Integer.parseInt(preLatUpd),1000000,6).doubleValue();
        double preLon = BigDecimalUtils.divide(StringUtils.isEmpty(preLonUpd)?pre.getOriginalLng():Integer.parseInt(preLonUpd),1000000,6).doubleValue();
        return AccumulatorUtils.getLatLonDiffence(curLat,curLon,preLat,preLon);
    }
    private void checkFirstPoint (List<Point0200> points , int sampleNum){
        if(CommonUtils.isCollectionEmpty(points)) return;
        List<DataNode> dpoints = new ArrayList<>();
        //int num = Math.max(points.size(),sampleNum);
        //取出样本数据
        Point0200 pre = null;
        int nodename = 0;
        for(int i=0;i<sampleNum;i++){
            Point0200 p = points.get(i);
            //如果当前点未移动，则过滤
            if(pre != null) {
                if (p.getSpeedKmh() == 0 && p.getOriginalLat() == pre.getOriginalLat() && p.getOriginalLng() == pre.getOriginalLng()) {
                    continue;
                }
            }
            double[] latlon = new double[]{p.getOriginalLng(),p.getOriginalLat()};
            dpoints.add(new DataNode(nodename,latlon,i));
            nodename++;
            pre = p;
        }
        if(dpoints.size() >= 50) {
            //计算样本中每个点局部离群点因子
            List<DataNode> resllt = OutlierNodeDetect.getOutlierNode(dpoints);
            //找到第一个局部离群点因子在范围内的点，重新计算该点之前的经纬度坐标
            updFirstPoint(points, resllt);
        }else{
            logger.error("样本数量小于50,不符合要求，不执行LOF算法。tid:{},上传点数量:{},样本点数量:{}",points.get(0).getTid(),points.size(),dpoints.size());
        }
    }

    private void updFirstPoint(List<Point0200> points, List<DataNode> nodes){
        if(CommonUtils.isCollectionEmpty(nodes))return;
        //如果第一个点未发生漂移则不需要修改坐标
        if(nodes.get(0).getLof() <= 2)return;
        for(int i=0;i<nodes.size();i++){
            if(nodes.get(i).getLof() > 2){
                //如果发生漂移，则跳过
                continue;
            }else {
                //找到第一个未发生漂移的位置点，修正该点之前的经纬度坐标
                //修正第一个正常点之前的经纬度
                //得到过滤停车点前位置点序号
                int pointnum = nodes.get(i).getPointNum();
                Point0200 p0 = points.get(pointnum);

                //计算前两个正常点的方向
                double direction = p0.getDirection();
                Point0200 pre= p0;
                //处理之前错误的经纬度
                for (int j=pointnum-1 ;j>=0;j--){
                    int newLat = 0;
                    int newLon = 0;
                    int preLat = AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT)==null?
                            pre.getOriginalLat():
                            Integer.parseInt(AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLAT));
                    int preLon = AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG)==null?
                            pre.getOriginalLng():
                            Integer.parseInt(AccumulatorUtils.getFieldFromConcatString(pre.getDataUpd(),"\\|",DCConstants.UPDFIELD_0200_ORIGINALLNG));
                    Point0200 cur = points.get(j);
                    if(pre.getSpeedKmh()+cur.getSpeedKmh() == 0 || pre.getGpsTime() == cur.getGpsTime()){
                        newLat = preLat;
                        newLon = preLon;
                    }else{
                        //距离,单位：km
                        double diff_mileage = abs(cur.getStandardMileage() - pre.getStandardMileage());
                        double diff_gpstime = cur.getGpsTime() == pre.getGpsTime()? 0.0: BigDecimalUtils.divide((cur.getGpsTime() - pre.getGpsTime()),1000,3).doubleValue();
                        double distance = (3600*diff_mileage/diff_gpstime) <= Speed_Max_Threshold? diff_mileage
                                :BigDecimalUtils.divide(diff_gpstime*(cur.getSpeedKmh()+pre.getSpeedKmh()),7200,3).doubleValue();
                        Tuple2<Double,Double> newLatLon =
                                AccumulatorUtils.getLatLon(preLat*Math.pow(10,-6), preLon*Math.pow(10,-6),
                                        distance,direction);
                        //计算出前一点的经纬度坐标
                        newLat = BigDecimalUtils.multiply(newLatLon._1,1000000,0).intValue();
                        newLon = BigDecimalUtils.multiply(newLatLon._2,1000000,0).intValue();
                    }
                    //更新坐标
                    //修正结果记录到修正值列
                    String updString = DCConstants.UPDFIELD_0F37_ORIGINALLAT+"="+newLat+"|"+DCConstants.UPDFIELD_0F37_ORIGINALLNG+"="+newLon;
                    //String updString = DCConstants.UPDFIELD_0F37_ORIGINALLAT+"="+points.get(j).getOriginalLat()+"|"+DCConstants.UPDFIELD_0F37_ORIGINALLNG+"="+points.get(j).getOriginalLng();
                    Point0200 upd = points.get(j);
                    upd.setDataUpd(StringUtils.isEmpty(upd.getDataUpd())?updString :upd.getDataUpd()+"|"+updString);
                    //更新原始点坐标
                    //upd.setOriginalLat(newLat);
                    //upd.setOriginalLng(newLon);
                    ++abnormal_lat_lon_count;
                    pre = cur;
                }
                break;
            }

        }
    }


}
